package xw4g08.sameAs;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import org.apache.commons.math.distribution.ZipfDistributionImpl;
import org.openrdf.model.Resource;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.model.ValueFactory;
import org.openrdf.model.impl.ValueFactoryImpl;
import org.openrdf.rio.RDFFormat;
import org.openrdf.rio.RDFHandler;
import org.openrdf.rio.RDFHandlerException;
import org.openrdf.rio.Rio;
import org.openrdf.rio.helpers.RDFHandlerBase;

import xw4g08.Configuration;

public class SAGenHandler extends RDFHandlerBase {

	private final double resRate = 4174654.0/2000000000;//the ratio of resources having sameAs
	private final double exp = 2.528;//the exponent of the power law
	
	private Set<Resource> nodes = new HashSet<Resource>();//records all nodes
	private Map<String,List<Resource>> categories = new HashMap<String, List<Resource>>();//keeps all nodes in different types
	private final String[] types = {"ProductFeature","ProductType","Product","Producer","Offer","Vendor","Reviewer","Review"};//the order of types is crucial for correct matching
	private long statementCounter = 0;
	private long sacount = 0;
	private RDFFormat format = RDFFormat.TURTLE;
	
	private String filename;
	private RDFHandler writer;
	private ValueFactory vf;
	private URI sameAs;
	
	/**
	 * Path of output file will be decided automatically
	 * @param filename The name of the output file.
	 */
	public SAGenHandler(String filename) {
		this.filename = filename;
		vf = ValueFactoryImpl.getInstance();
		sameAs = vf.createURI("http://www.w3.org/2002/07/owl#","sameAs");
	}
	
	public void SetFormat(RDFFormat format) {
		this.format = format;
	}
	
	@Override
	public void startRDF()
			throws RDFHandlerException {
		System.out.println("SameAs handeler started.");
		FileOutputStream out;
		try {
			out = new FileOutputStream(new File(Configuration.OUT_DIR, filename));
			writer = Rio.createWriter(format, out);
			writer.startRDF();
			writer.handleNamespace("owl", "http://www.w3.org/2002/07/owl#");
			
			for(String type:types) {
				categories.put(type, new ArrayList<Resource>());
			}
			categories.put("General", new ArrayList<Resource>());//for those don't have a type
			
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		}
		
		
	}

	
	@Override
	public void handleNamespace(String prefix, String uri)
			throws RDFHandlerException {
		writer.handleNamespace(prefix, uri);
	}

	@Override
	/**
	 * The purpose of this procedure is to record all nodes of input statements.
	 * Co-reference is generated in {@code endRDF()} when all nodes have been recorded.
	 */
	public void handleStatement(Statement st)
		throws RDFHandlerException {
		
		Value obj = st.getObject();
		Value sub = st.getSubject();
		
		if(obj instanceof Resource) {
			Resource temp = (Resource)obj;
			nodes.add(temp);
			categories.get(typeOf(temp)).add(temp);
		}
		if(sub instanceof Resource) {
			Resource temp = (Resource) sub;
			nodes.add(temp);
			categories.get(typeOf(temp)).add(temp);
		}
		statementCounter++;
		if(statementCounter%100000 == 0) {
			System.out.println(statementCounter/1000000.0 + "M triples processed.");
		}
	}
	
	/**
	 * decide the type of a resource according to the last part of its URI
	 * e.g. the type of <http://www4.xxx/dataFromProducer9685/Product491100> is Product.
	 * @param res
	 * @return the type of {@code res}
	 */
	private String typeOf(Resource res) {
		
		String[] parts = res.stringValue().split("/");
		String last = parts[parts.length-1];
		
		for(String type:types) {
			if(last.contains(type))
				return type;
		}
		
		return "General";
	}
	
	@Override
	public void endRDF()
		throws RDFHandlerException {
		System.out.println("Generating co-references.");
		Random ran = new Random();
		int size = nodes.size();
		int n = (int) (size*resRate);//the number of nodes having sameAs URIs
		if(n<10)
			n = 10;//if the number is too small, set to 10
		ZipfDistributionImpl zipf = new ZipfDistributionImpl(n, exp);
		List<Resource> nodelist = new ArrayList<Resource>();
		nodelist.addAll(nodes);
		//Randomly select n nodes out of all the nodes as those having sameAs URIs. 
		//For each of them, generate m sameAs URIs of it according to Zipf's law, 
		//and then randomly select m nodes to generate sameAs statements.
		for(int i=0;i<n;i++) {
			int subInx = ran.nextInt(size);//the index of the subject of co-reference
			Resource sub = nodelist.get(subInx);
			int m = GeneratorCore.nextPow(zipf);//the number of co-reference of this subject
			List<Resource> category = categories.get(typeOf(sub));
			if(m > category.size())
				m = category.size();
			
			for(int j=0;j<m;j++) {
				//if there's only one node and it's the same as the subject, jump to the next subject
				if(category.size()==1 && sub.equals(category.get(0)))
					continue;
				
				Resource obj;
				while(true) {
					int objInx = ran.nextInt(category.size());
					obj = category.get(objInx);
					if(!sub.equals(obj))
						break;
					
				}
				
				Statement sta = vf.createStatement(sub, sameAs, obj);
				writer.handleStatement(sta);
				sta = vf.createStatement(obj, sameAs, sub);
				writer.handleStatement(sta);
				sacount++;
				
			}
		}
		writer.endRDF();
		System.out.println(sacount + " co-references generated.");
	}
	
	
	
}
